#include <string>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>

#include "cnats.h"
#include "clog.h"
#include "threads.h"

using namespace std;

enum STATE{
    WAIT_CONTROL,
    WAIT_DATA
};


class asyncCaller: public worker {

public:
    asyncCaller();
    void addEvent();
    void run(){
    }
};


// this Thread receive and store caller's request and send asyn,
// reply from nats server will be processd async in other thread.
class ReqProcessor: public worker{
private:
    cnats * parent;
public:
    ReqProcessor(cnats * conn){
        parent = conn;
    }
    void run(){
        clog::dl("ReqProcessor start");
        static char localBuf[1024];
        int len;
        while(true){
            sem_wait(&parent->ReqSem);
            sem_wait(&parent->SubPubMutex);
                if(parent->ReqMessages.size() <= 0){
                    sem_post(&parent->SubPubMutex);
                    break;
                }
                msgsToSend *msg = parent->ReqMessages.front();
                parent->ReqMessages.pop_front();
            sem_post(&parent->SubPubMutex);
            int retlen;
            clog::dl("process msg");
            switch(msg->type){
            case TYPE_SUB:
                if (msg->subj.length() == 0 || msg->handler == NULL) {
                    clog::el("Illegial Subj");
                    break;
                }
                clog::d("SUNSCRRIBE Subj:");
                clog::dl(msg->subj);
                parent->SSID++;
                if (parent->subscribeMap.find(msg->subj) != parent->subscribeMap.end() ){
                    break;
                }
                len = sprintf(localBuf, "SUB %s %d\r\n", msg->subj.data(), parent->SSID);
                parent->subscribeMap[msg->subj] = msg->handler;
                retlen = send(parent->socketfd,localBuf,len,0);
                if(retlen != len){
                    clog::el("ERROR send");
                    parent->issueReconnect();
                    break;
                }
                sem_wait(&parent->SubPubMutex);
                    parent->ProcessedMessages.push_back(msg);
                sem_post(&parent->SubPubMutex);
                break;
            case TYPE_PUB:
                if(msg->subj.length() == 0){
                    clog::el("Illegial Subj");
                    break;
                }
                clog::d("PUNLISH Subj:");
                clog::dl(msg->subj);
                len = sprintf(localBuf, "PUB %s %d\r\n", msg->subj.data(),msg->dataLen);
                retlen = send(parent->socketfd,localBuf,len,0);
                if(retlen != len){
                    clog::el("ERROR Send PUB message");
                    parent->issueReconnect();
                }
                retlen = send(parent->socketfd, msg->data, msg->dataLen, 0);
                if(retlen != msg->dataLen){
                    clog::el("ERROR Send PUB message load");
                }
                if (send(parent->socketfd,"\r\n",2,0) != 2){
                    clog::el("ERROR send PUB message \\r\\n");
                }
                sem_wait(&parent->SubPubMutex);
                    parent->ProcessedMessages.push_back(msg);
                sem_post(&parent->SubPubMutex);
                break;
            default:
                clog::el("ERROR message Type.");
                break;
            }
        }
    }
    void stop(){
        sem_post(&parent->ReqSem);
    }
};

// this thread recv data from socket and process data.
// process data are stored in a queue, and corresponding
// functions should be called async in another thread.
class MsgProcessor: public worker{
private:
    // 16MB buffer size
    static const int BUFFER_SIZE = 16 * 1024 * 1024;
    bool    cont;
    cnats*  parent;
    char*   buffer;
    int     bufferLen;
    int     bufferStart;
    int     state;
    int     loadSize;
    string  reply;
    string  subject;

    void split(string& s, string delim, vector< string >* ret)
    {
        size_t last = 0;
        size_t index=s.find_first_of(delim,last);
        while (index!=string::npos)
        {
            ret->push_back(s.substr(last,index-last));
            last=index+1;
            index=s.find_first_of(delim,last);
        }
        if (index-last>0)
        {
            ret->push_back(s.substr(last,index-last));
        }
    }
    void getData(){
        if(BUFFER_SIZE == bufferLen){
            memmove(buffer, bufferStart+buffer, bufferLen-bufferStart);
            clog::d("Copy Data :");
            clog::dl(bufferLen - bufferStart);
            bufferLen = bufferLen - bufferStart;
            bufferStart = 0;
        }
        int len = 0;
        len = recv(parent->socketfd, buffer + bufferLen, BUFFER_SIZE - bufferLen,0);
        if (len < 0) {
            parent->issueReconnect();
            return;
        }
        bufferLen += len;
    }

    bool processCtl(){
        int index;
        bool succ = false;
        for (index = bufferStart; index < bufferLen - 1; index++) {
            if (buffer[index] == '\r' && buffer[index + 1] == '\n') {
                succ = true;
                break;
            }
        }
        if (!succ) {
            return false;
        }
        string str(buffer + bufferStart, index - bufferStart);
        vector<string> strs;
        split(str, string(" "), &strs);
        clog::d("PARSED CTL :[");
        for (unsigned int i = 0; i < strs.size(); i++) {
            clog::d(strs[i]);
            clog::d(" , ");
        }
        clog::dl("]");
        if (strs.size() == 4) {
            subject = strs[1];
            loadSize = atoi(strs[3].data());
        } else if (strs.size() == 5){
            subject = strs[1];
            reply = strs[3];
            loadSize = atoi(strs[4].data());
        }
        bufferStart = index + 2;
        return true;
    }
    bool processPing(){
        clog::dl("Process Ping Msg");
        int index;
        bool succ = false;
        for (index = bufferStart; index < bufferLen - 1; index++) {
            if (buffer[index] == '\r' && buffer[index + 1] == '\n') {
                succ = true;
                break;
            }
        }
        if (!succ) {
            return false;
        }
        int len = send(parent->socketfd, "PONG\r\n", 6, 0);
        if(len != 6){
            parent->issueReconnect();
        }
        bufferStart = index + 2;
        return true;
    }
    bool processReply(){
        int index;
        bool succ = false;
        for (index = bufferStart; index < bufferLen - 1; index++) {
            if (buffer[index] == '\r' && buffer[index + 1] == '\n') {
                succ = true;
                break;
            }
        }
        if (!succ) {
            return false;
        }
        clog::d("REPLY:");
        clog::dl(string(buffer + bufferStart,index - bufferStart));
        if(parent->ProcessedMessages.size()){
            sem_wait(&parent->SubPubMutex);
                msgsToSend * msg = parent->ProcessedMessages.front();
                parent->ProcessedMessages.pop_front();
            sem_post(&parent->SubPubMutex);
            // TODO move this part ,using a queue to call these funcs
            // if(msg->handler){
            //     if(strncmp(buffer + bufferStart, "+OK", 3) == 0){
            //         msg->handler->asyncSuccCallBack();
            //     }else{
            //         msg->handler->asyncErrorCallBack();
            //     }
            // }
            delete msg;
        }
        bufferStart = index + 2;
        return true;
    }
    bool processMsg(){
        // recvedMsgs *msg = new recvedMsgs(subject,reply,new char[loadSize+2],loadSize);
        recvedMsgs msg(subject,reply,buffer+bufferStart,loadSize);
        // memcpy(msg->data, buffer+bufferStart, loadSize);
        // msg->data[loadSize] = '\r';
        // msg->data[loadSize+1] = '\n';
        // TODO make a queue and process in another queue
        if (parent->subscribeMap.find(subject) != parent->subscribeMap.end()) {
            parent->subscribeMap[subject]->execute(&msg);
        }
        bufferStart += loadSize + 2;
        return true;
    }

public:
    MsgProcessor(cnats *p){
        cont = true;
        parent = p;
        buffer = new char[BUFFER_SIZE];
        if(buffer == NULL){
            clog::el("BUFFER ALLOC ERROR");
        }
        bufferLen = 0;
        bufferStart = 0;
        state = WAIT_CONTROL;
    }

    void run(void){
        clog::dl("MsgProcess Start");
        while (cont) {
            if (parent->reconnectFlag) {
                parent->reconnect();
                bufferLen = 0;
                bufferStart = 0;
                state = WAIT_CONTROL;
                loadSize = 0;
            }
            if (bufferLen - bufferStart < 6) {
                getData();
            }
            string cmd(buffer+bufferStart, 6);
            switch (state) {
                case WAIT_CONTROL:
                    if (cmd.compare(0, 3, "MSG")==0) {
                        if (!processCtl()){
                            getData();
                        } else {
                            state = WAIT_DATA;
                        }
                        break;
                    }
                    if (cmd.compare(0,4,"PING")==0) {
                        processPing();
                        break;
                    }
                    if (cmd.compare(0,1,"+")==0 || cmd.compare(0,1,"-")==0){
                        if(!processReply()){
                            getData();
                        }
                        break;
                    }
                    clog::el("Not Handled Cmd:" + cmd);
                    break;
                case WAIT_DATA:
                    if (loadSize + 2 > bufferLen - bufferStart) {
                        getData();
                    } else {
                        clog::dl("Get Message");
                        processMsg();
                        state = WAIT_CONTROL;
                    }
                    break;
            }
        }
    }

    void stop(void){
        clog::dl("Processor Stop");
        cont = false;
    }
};

cnats::cnats():threadPool(0){
    socketfd = 0;
    serverCurrent = -1;
    reconnectFlag = true;
    unsigned int err = sem_init(&SubPubMutex,0,1);
    err |= sem_init(&ReqSem,0,0);
    if(err){
        clog::el("ERROR Init SubPubMutex");
    }
}

Server::Server(string host,int port){
    if(host.compare("localhost")==0){
        host = string("127.0.0.1");
    }
    this->host = host;
    this->port = port;
}

cnats* cnats::Connect(string urls){
    int pos = 0;
    bool more = true;
    cnats *ret = new cnats;
    while(more){
        int epos = urls.find(",", pos);
        if(epos < 0){
            more = false;
            epos = urls.length();
        }
        string url = urls.substr(pos, epos - pos);
        clog::d("PARSE URL:");
        clog::dl(url);
        if(url.compare(0,7,"nats://")){
            clog::e("ERROR URL:");
            clog::el(url);
        }
        url = url.substr(7);
        int index = url.find(":");
        int port;
        string host;
        if(index < 0){
            port = 4222;
        } else {
            port = atoi(url.substr(index+1).data());
        }
        host = url.substr(0,index);
        clog::d("PARSED: port:");
        clog::d(port);
        clog::d(" host:");
        clog::dl(host);
        ret->addServer(host,port);
        pos = epos+1;
    }
    ret->reconnect();
    ret->threadPool.addJob(new MsgProcessor(ret));
    ret->threadPool.addJob(new ReqProcessor(ret));
    return ret;
}

bool cnats::addServer(string host, int port){
    Server s(host,port);
    this->servers.push_back(s);
    clog::d("ADDING SERVER:");
    clog::d(host);
    clog::d(":");
    clog::dl(port);
    return true;
}

void cnats::reconnect(){
    struct sockaddr_in addr;
    char cmdbuffer[2048];
    int len;
    clog::dl("Reconnecting ...");
    if(socketfd){
        close(socketfd);
    }
    while(reconnectFlag){
        serverCurrent = (serverCurrent + 1) % (int)servers.size();
        clog::d("Trying to connect : ");
        clog::d(servers[serverCurrent].host);
        clog::d(":");
        clog::dl(servers[serverCurrent].port);
        memset(&addr,0,sizeof(addr));
        addr.sin_family = AF_INET;
        addr.sin_addr.s_addr = inet_addr(servers[serverCurrent].host.data());
        addr.sin_port = htons(servers[serverCurrent].port);
        socketfd = socket(AF_INET,SOCK_STREAM,0);
        if(socketfd <= 0){
            clog::dl("ERROR socket");
            continue;
        }
        if(connect(socketfd,(struct sockaddr*)&addr,sizeof(addr))==-1){
            clog::dl("ERROR connect");
            continue;
        }
        len = recv(socketfd,cmdbuffer,sizeof(cmdbuffer),0);
        if(len>2){
            cmdbuffer[len-2] = 0;
        }
        cmdbuffer[len] = 0;
        clog::d("Recvd Msg:");
        clog::dl(cmdbuffer);
        
        reconnectFlag = false;
    }
}

void cnats::issueReconnect(){
    reconnectFlag = true;
}

bool cnats::Subscribe(string subj, MsgHandler *handler) {
    sem_wait(&SubPubMutex);
        ReqMessages.push_back(new msgsToSend(subj, "",handler));
    sem_post(&SubPubMutex);
    sem_post(&ReqSem);
    return true;
}

bool cnats::Publish(string subj, const char * data, int len, MsgHandler* handler){
    sem_wait(&SubPubMutex);
        ReqMessages.push_back(new msgsToSend(subj,data,len,handler));
    sem_post(&SubPubMutex);
    sem_post(&ReqSem);
    return true;
}
